#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
StarRocks操作模块
提供事务查询、回滚等数据库操作功能
"""

import json
import logging
import subprocess
import requests
import time
from datetime import datetime, timedelta
from typing import Optional, Dict, Any, List

import pymysql
from flask import Blueprint, request, jsonify

from app.utils.logger import log_api_call, log_database_operation, log_business_logic
from app.utils.mysql_db import get_db_connection

# 创建Blueprint
starrocks_bp = Blueprint('starrocks', __name__, url_prefix='/api/starrocks')

# 配置日志
logger = logging.getLogger(__name__)

def success_response(data: Any = None, message: str = "操作成功") -> Dict:
    """统一成功响应格式"""
    return {
        "success": True,
        "message": message,
        "data": data,
        "timestamp": datetime.now().isoformat()
    }

def error_response(message: str, error_code: str = "UNKNOWN_ERROR") -> Dict:
    """统一错误响应格式"""
    return {
        "success": False,
        "error_code": error_code,
        "message": message,
        "timestamp": datetime.now().isoformat()
    }

@log_database_operation
def get_starrocks_configs() -> List[Dict]:
    """获取StarRocks配置列表"""
    try:
        with get_db_connection() as conn:
            cursor = conn.cursor(pymysql.cursors.DictCursor)
            
            sql = """
                SELECT id, config_name, host, port, username, password, 
                       `database`, extra_config, description
                FROM service_config 
                WHERE service_type = 'starrocks' AND is_deleted = 0 AND is_active = 1
                ORDER BY sort_order
            """
            cursor.execute(sql)
            configs = cursor.fetchall()
            
            # 解析extra_config
            for config in configs:
                if config['extra_config']:
                    extra = json.loads(config['extra_config'])
                    config.update(extra)
            
            return configs
            
    except Exception as e:
        logger.error(f"获取StarRocks配置失败: {e}")
        raise

@log_database_operation
def get_starrocks_config_by_id(config_id: int) -> Dict:
    """根据ID获取StarRocks配置"""
    try:
        with get_db_connection() as conn:
            cursor = conn.cursor(pymysql.cursors.DictCursor)
            
            sql = """
                SELECT id, config_name, host, port, username, password, 
                       `database`, extra_config
                FROM service_config 
                WHERE id = %s AND service_type = 'starrocks' AND is_deleted = 0
            """
            cursor.execute(sql, (config_id,))
            config = cursor.fetchone()
            
            if not config:
                raise ValueError(f"StarRocks配置 {config_id} 不存在")
                
            # 解析extra_config
            if config['extra_config']:
                extra = json.loads(config['extra_config'])
                config.update(extra)
            
            return config
            
    except Exception as e:
        logger.error(f"获取StarRocks配置失败: {e}")
        raise

def create_starrocks_connection(config: Dict) -> pymysql.Connection:
    """创建StarRocks数据库连接（带重试机制）"""
    connection_params = {
        'host': config['host'],
        'port': config['port'],
        'user': config['username'],
        'password': config['password'],
        'database': config.get('database', ''),
        'charset': config.get('charset', 'utf8mb4'),
        'autocommit': config.get('autocommit', True),
        'connect_timeout': config.get('connect_timeout', 10),
        'read_timeout': config.get('read_timeout', 30),
        'write_timeout': config.get('write_timeout', 30)
    }
    
    # 添加重试逻辑（与其他连接保持一致）
    max_retries = 3
    last_error = None
    
    for retry in range(max_retries):
        try:
            if retry > 0:
                logger.info(f"🔄 重试StarRocks连接 (尝试{retry+1}/{max_retries}) [{config['host']}:{config['port']}]")
            
            conn = pymysql.connect(**connection_params)
            logger.info(f"StarRocks连接创建成功: {config['host']}:{config['port']}")
            return conn
            
        except Exception as e:
            last_error = e
            logger.error(f"创建StarRocks连接失败 (尝试{retry+1}/{max_retries}): {e}")
            
            if retry < max_retries - 1:
                # 递增等待：第1次失败等30秒，第2次失败等60秒
                wait_time = 30 if retry == 0 else 60
                logger.info(f"⏰ 等待{wait_time}秒后重试...")
                time.sleep(wait_time)
            else:
                # 3次都失败了
                logger.error(f"❌ StarRocks连接重试{max_retries}次均失败")
    
    # 最终失败，抛出最后的异常
    raise last_error

@log_business_logic("query_running_transactions")
def query_running_transactions(config: Dict, database: Optional[str] = None) -> List[Dict]:
    """查询运行中的事务"""
    conn = None
    try:
        conn = create_starrocks_connection(config)
        cursor = conn.cursor(pymysql.cursors.DictCursor)
        
        # 查询运行中的事务
        if database:
            sql = """
                SELECT 
                    TXN_ID as transaction_id,
                    LABEL as label,
                    DATABASE_NAME as database_name,
                    STATE as state,
                    PREPARE_TIME as prepare_time,
                    PRE_COMMIT_TIME as pre_commit_time,
                    COMMIT_TIME as commit_time,
                    PUBLISH_TIME as publish_time,
                    FINISH_TIME as finish_time,
                    REASON as reason,
                    ERROR_REPLICAS as error_replicas,
                    TIMEOUT_MS as timeout_ms
                FROM information_schema.transactions 
                WHERE DATABASE_NAME = %s 
                AND STATE IN ('PREPARE', 'PRECOMMITTED', 'COMMITTED', 'VISIBLE')
                ORDER BY PREPARE_TIME DESC
            """
            cursor.execute(sql, (database,))
        else:
            sql = """
                SELECT 
                    TXN_ID as transaction_id,
                    LABEL as label,
                    DATABASE_NAME as database_name,
                    STATE as state,
                    PREPARE_TIME as prepare_time,
                    PRE_COMMIT_TIME as pre_commit_time,
                    COMMIT_TIME as commit_time,
                    PUBLISH_TIME as publish_time,
                    FINISH_TIME as finish_time,
                    REASON as reason,
                    ERROR_REPLICAS as error_replicas,
                    TIMEOUT_MS as timeout_ms
                FROM information_schema.transactions 
                WHERE STATE IN ('PREPARE', 'PRECOMMITTED', 'COMMITTED', 'VISIBLE')
                ORDER BY PREPARE_TIME DESC
                LIMIT 100
            """
            cursor.execute(sql)
        
        transactions = cursor.fetchall()
        
        # 转换时间格式
        for txn in transactions:
            for time_field in ['prepare_time', 'pre_commit_time', 'commit_time', 'publish_time', 'finish_time']:
                if txn.get(time_field):
                    txn[time_field] = str(txn[time_field])
        
        return transactions
        
    except Exception as e:
        logger.error(f"查询运行中事务失败: {e}")
        raise
    finally:
        if conn:
            conn.close()

@log_business_logic("query_running_transactions_by_show_proc")
def query_running_transactions_by_show_proc(config: Dict, database: str) -> List[Dict]:
    """使用SHOW PROC命令查询运行中的事务（兼容脚本方式）"""
    conn = None
    try:
        conn = create_starrocks_connection(config)
        cursor = conn.cursor(pymysql.cursors.DictCursor)
        
        # 使用与脚本相同的SHOW PROC命令
        sql = f"SHOW PROC '/transactions/{database}/running'"
        cursor.execute(sql)
        
        transactions = cursor.fetchall()
        logger.info(f"通过SHOW PROC查询到 {len(transactions)} 个运行中的事务")
        
        return transactions
        
    except Exception as e:
        logger.error(f"通过SHOW PROC查询运行中的事务失败: {e}")
        raise

    finally:
        if conn:
            conn.close()

@log_business_logic("filter_transactions_by_time")
def filter_transactions_by_time(transactions: List[Dict], time_condition: str = None) -> List[Dict]:
    """
    根据PrepareTime条件过滤事务，支持多种时间条件
    """
    if not time_condition:
        return transactions

    now = datetime.now()
    filtered = []

    # 预解析时间条件
    cond = time_condition.strip()
    is_range = ',' in cond
    is_before = cond.startswith('<')
    is_after = cond.startswith('>')
    presets = {'today', 'yesterday', 'last_hour', 'last_2_hours', 'last_6_hours', 'last_12_hours', 'last_24_hours'}

    eq_time = None
    start_t = end_t = None
    before_t = after_t = None
    fmt = '%Y-%m-%d %H:%M:%S'

    try:
        if is_range:
            start_s, end_s = cond.split(',', 1)
            start_t = datetime.strptime(start_s.strip(), fmt)
            end_t = datetime.strptime(end_s.strip(), fmt)
        elif is_before:
            before_t = datetime.strptime(cond[1:].strip(), fmt)
        elif is_after:
            after_t = datetime.strptime(cond[1:].strip(), fmt)
        elif cond not in presets:
            # 等于指定时间
            eq_time = datetime.strptime(cond, fmt)
    except Exception as e:
        logger.warning(f"无效的时间格式: {cond}, 错误: {e}")
        return []

    for txn in transactions:
        # 适配不同的时间字段名
        pt_str = txn.get('PrepareTime') or txn.get('prepare_time')
        if not pt_str:
            continue
        
        try:
            prepare_time = datetime.strptime(str(pt_str), fmt)
        except Exception as e:
            logger.warning(f"无法解析PrepareTime: {pt_str} 错误: {e}")
            continue

        # 应用条件
        if cond == 'today':
            if prepare_time.date() == now.date():
                filtered.append(txn)
        elif cond == 'yesterday':
            if prepare_time.date() == (now - timedelta(days=1)).date():
                filtered.append(txn)
        elif cond == 'last_hour':
            if prepare_time >= now - timedelta(hours=1):
                filtered.append(txn)
        elif cond == 'last_2_hours':
            if prepare_time >= now - timedelta(hours=2):
                filtered.append(txn)
        elif cond == 'last_6_hours':
            if prepare_time >= now - timedelta(hours=6):
                filtered.append(txn)
        elif cond == 'last_12_hours':
            if prepare_time >= now - timedelta(hours=12):
                filtered.append(txn)
        elif cond == 'last_24_hours':
            if prepare_time >= now - timedelta(hours=24):
                filtered.append(txn)
        elif is_range:
            if start_t <= prepare_time <= end_t:
                filtered.append(txn)
        elif is_before:
            if prepare_time < before_t:
                filtered.append(txn)
        elif is_after:
            if prepare_time > after_t:
                filtered.append(txn)
        else:
            # 等于指定时间
            if eq_time and prepare_time == eq_time:
                filtered.append(txn)

    return filtered

@log_business_logic("query_databases")
def query_databases(config: Dict) -> List[Dict]:
    """查询StarRocks数据库列表"""
    conn = None
    try:
        conn = create_starrocks_connection(config)
        cursor = conn.cursor(pymysql.cursors.DictCursor)
        
        # 查询数据库列表，先尝试information_schema，如果失败则用SHOW DATABASES
        try:
            sql = "SELECT * FROM information_schema.SCHEMATA ORDER BY SCHEMA_NAME"
            cursor.execute(sql)
            databases = cursor.fetchall()
            logger.info(f"通过information_schema查询到 {len(databases)} 个数据库")
        except Exception as e:
            logger.warning(f"information_schema查询失败: {e}，尝试使用SHOW DATABASES")
            sql = "SHOW DATABASES"
            cursor.execute(sql)
            raw_databases = cursor.fetchall()
            # 转换SHOW DATABASES的结果为标准格式
            databases = []
            for db in raw_databases:
                # SHOW DATABASES返回的格式可能是{'Database': 'db_name'}
                db_name = db.get('Database') or list(db.values())[0] if db else None
                if db_name:
                    databases.append({
                        'SCHEMA_NAME': str(db_name),
                        'CATALOG_NAME': 'default_catalog',
                        'DEFAULT_CHARACTER_SET_NAME': 'utf8'
                    })
            logger.info(f"通过SHOW DATABASES查询到 {len(databases)} 个数据库")
        
        # 系统数据库列表，需要排除
        system_schemas = {
            'information_schema',
            '_statistics_',
            'sys',
            'starrocks_audit_db__'
        }
        
        # 转换为简化格式并过滤系统库
        result = []
        for db in databases:
            schema_name = db.get('SCHEMA_NAME') or db.get('schema_name')
            
            # 排除系统数据库
            if schema_name and schema_name.lower() not in system_schemas:
                # 确保所有值都是字符串类型，避免序列化问题
                catalog_name = db.get('CATALOG_NAME') or db.get('catalog_name') or ''
                charset_name = db.get('DEFAULT_CHARACTER_SET_NAME') or db.get('default_character_set_name') or ''
                
                result.append({
                    'schema_name': str(schema_name) if schema_name else '',
                    'catalog_name': str(catalog_name) if catalog_name else '',
                    'default_character_set_name': str(charset_name) if charset_name else ''
                })
        
        logger.info(f"过滤系统库后剩余 {len(result)} 个业务数据库")
        return result
        
    except Exception as e:
        logger.error(f"查询数据库列表失败: {e}")
        raise

    finally:
        if conn:
            conn.close()

@log_business_logic("rollback_transaction_by_label")
def rollback_transaction_by_label(config: Dict, label: str, database: str) -> Dict:
    """
    根据label回滚事务（使用HTTP API方式，兼容脚本）
    """
    try:
        # 构建回滚API请求 (使用8060端口的HTTP API)
        rollback_url = f"http://{config['host']}:8060/api/transaction/rollback"
        
        headers = {
            'label': label,
            'Expect': '100-continue',
            'db': database
        }
        
        # 使用rollback专用的认证信息，如果没有则使用普通认证
        rollback_user = config.get('rollback_user', config['username'])
        rollback_password = config.get('rollback_password', config['password'])
        auth = (rollback_user, rollback_password)
        
        logger.info(f"执行rollback: label={label}, database={database}")
        
        response = requests.post(
            rollback_url,
            headers=headers,
            auth=auth,
            timeout=30
        )
        
        if response.status_code == 200:
            logger.info(f"Rollback成功: {label}")
            return {
                "status": "success",
                "message": "事务回滚成功",
                "label": label
            }
        else:
            error_msg = f"Rollback失败: {label}, HTTP状态码: {response.status_code}, 响应: {response.text}"
            logger.error(error_msg)
            return {
                "status": "failed",
                "message": error_msg,
                "label": label
            }
            
    except Exception as e:
        error_msg = f"Rollback异常: {label}, 错误: {str(e)}"
        logger.error(error_msg)
        return {
            "status": "failed",
            "message": error_msg,
            "label": label
        }

@log_business_logic("rollback_transactions")
def rollback_transactions(config: Dict, transaction_ids: List[int], database: Optional[str] = None) -> Dict:
    """批量回滚事务"""
    conn = None
    results = {
        "success_count": 0,
        "failed_count": 0,
        "results": []
    }
    
    try:
        conn = create_starrocks_connection(config)
        cursor = conn.cursor()
        
        for txn_id in transaction_ids:
            try:
                # 构建回滚SQL
                if database:
                    sql = f"CANCEL LOAD FROM {database} WHERE LABEL = (SELECT LABEL FROM information_schema.transactions WHERE TXN_ID = {txn_id})"
                else:
                    sql = f"CANCEL LOAD WHERE LABEL = (SELECT LABEL FROM information_schema.transactions WHERE TXN_ID = {txn_id})"
                
                cursor.execute(sql)
                conn.commit()
                
                results["success_count"] += 1
                results["results"].append({
                    "transaction_id": txn_id,
                    "status": "success",
                    "message": "事务回滚成功"
                })
                
                logger.info(f"事务 {txn_id} 回滚成功")
                
            except Exception as e:
                results["failed_count"] += 1
                results["results"].append({
                    "transaction_id": txn_id,
                    "status": "failed",
                    "message": str(e)
                })
                
                logger.error(f"事务 {txn_id} 回滚失败: {e}")
        
        return results
        
    except Exception as e:
        logger.error(f"批量回滚事务失败: {e}")
        raise
    finally:
        if conn:
            conn.close()

@log_business_logic("execute_sql_query")
def execute_sql_query(config: Dict, sql: str, database: Optional[str] = None) -> Dict:
    """执行SQL查询"""
    conn = None
    try:
        conn = create_starrocks_connection(config)
        
        # 切换数据库
        if database:
            conn.select_db(database)
        
        cursor = conn.cursor(pymysql.cursors.DictCursor)
        
        # 记录查询开始时间
        start_time = datetime.now()
        
        # 执行查询
        cursor.execute(sql)
        
        # 获取结果
        if sql.strip().upper().startswith('SELECT') or sql.strip().upper().startswith('SHOW') or sql.strip().upper().startswith('DESCRIBE'):
            # 查询语句
            results = cursor.fetchall()
            row_count = len(results)
            
            # 限制返回行数（避免内存溢出）
            if row_count > 1000:
                results = results[:1000]
                truncated = True
            else:
                truncated = False
        else:
            # 非查询语句
            conn.commit()
            row_count = cursor.rowcount
            results = []
            truncated = False
        
        # 计算执行时间
        end_time = datetime.now()
        execution_time = (end_time - start_time).total_seconds()
        
        return {
            "results": results,
            "row_count": row_count,
            "execution_time_seconds": execution_time,
            "truncated": truncated,
            "sql": sql
        }
        
    except Exception as e:
        logger.error(f"执行SQL查询失败: {e}")
        raise
    finally:
        if conn:
            conn.close()


@log_business_logic("get_table_list")
def get_table_list(config: Dict, database: str) -> List[Dict]:
    """获取表列表"""
    conn = None
    try:
        conn = create_starrocks_connection(config)
        conn.select_db(database)
        cursor = conn.cursor(pymysql.cursors.DictCursor)
        
        # 获取表信息
        sql = """
            SELECT 
                TABLE_NAME as table_name,
                TABLE_TYPE as table_type,
                ENGINE as engine,
                TABLE_ROWS as table_rows,
                DATA_LENGTH as data_length,
                CREATE_TIME as create_time,
                UPDATE_TIME as update_time,
                TABLE_COMMENT as table_comment
            FROM information_schema.TABLES 
            WHERE TABLE_SCHEMA = %s
            ORDER BY TABLE_NAME
        """
        cursor.execute(sql, (database,))
        tables = cursor.fetchall()
        
        # 转换时间格式
        for table in tables:
            if table.get('create_time'):
                table['create_time'] = str(table['create_time'])
            if table.get('update_time'):
                table['update_time'] = str(table['update_time'])
        
        return tables
        
    except Exception as e:
        logger.error(f"获取表列表失败: {e}")
        raise
    finally:
        if conn:
            conn.close()


@log_business_logic("get_running_tasks")
def get_running_tasks(config: Dict, state_filter: str = None, sort_by: str = 'CREATE_TIME', sort_order: str = 'DESC') -> List[Dict]:
    """获取StarRocks任务列表，使用information_schema.task_runs表"""
    conn = None
    try:
        conn = create_starrocks_connection(config)
        cursor = conn.cursor(pymysql.cursors.DictCursor)
        
        # 构建查询SQL，使用information_schema.task_runs表
        # 检查StarRocks是否支持task_runs表，如果不支持则返回演示数据
        try:
            # 先测试表是否存在
            cursor.execute("SELECT COUNT(*) FROM information_schema.task_runs LIMIT 1")
            cursor.fetchone()
            
            # 如果表存在，构建查询SQL
            base_sql = """
            SELECT 
                QUERY_ID,
                TASK_NAME,
                CREATE_TIME,
                FINISH_TIME,
                STATE,
                CATALOG,
                `DATABASE`,
                DEFINITION,
                EXPIRE_TIME,
                ERROR_CODE,
                ERROR_MESSAGE,
                PROGRESS,
                EXTRA_MESSAGE,
                PROPERTIES
            FROM information_schema.task_runs
            """
        except Exception as table_check_error:
            logger.warning(f"task_runs表不存在或无法访问: {table_check_error}")
            # 直接返回空列表
            return []
        
        # 添加状态筛选条件
        where_conditions = []
        params = []
        
        if state_filter and state_filter.upper() != 'ALL':
            where_conditions.append("STATE = %s")
            params.append(state_filter.upper())
        
        # 构建完整的SQL
        if where_conditions:
            sql = base_sql + " WHERE " + " AND ".join(where_conditions)
        else:
            sql = base_sql
        
        # 添加排序
        valid_sort_columns = ['CREATE_TIME', 'FINISH_TIME', 'STATE', 'TASK_NAME', 'QUERY_ID']
        if sort_by.upper() in valid_sort_columns:
            sql += f" ORDER BY {sort_by} {sort_order.upper()}"
        else:
            sql += " ORDER BY CREATE_TIME DESC"  # 默认按创建时间降序
        
        # 限制返回数量，避免数据过多
        sql += " LIMIT 100"
        
        logger.info(f"执行任务查询SQL: {sql}")
        cursor.execute(sql, params)
        tasks = cursor.fetchall()
        
        # 格式化数据
        formatted_tasks = []
        for task in tasks:
            # 计算运行时长
            create_time = task.get('CREATE_TIME')
            finish_time = task.get('FINISH_TIME')
            duration_ms = 0
            
            if create_time:
                if finish_time:
                    # 已完成任务，计算总运行时长
                    if isinstance(create_time, str):
                        create_time = datetime.fromisoformat(create_time.replace('Z', '+00:00'))
                    if isinstance(finish_time, str):
                        finish_time = datetime.fromisoformat(finish_time.replace('Z', '+00:00'))
                    duration_ms = int((finish_time - create_time).total_seconds() * 1000)
                else:
                    # 运行中任务，计算当前运行时长
                    if isinstance(create_time, str):
                        create_time = datetime.fromisoformat(create_time.replace('Z', '+00:00'))
                    duration_ms = int((datetime.now() - create_time).total_seconds() * 1000)
            
            formatted_task = {
                'query_id': task.get('QUERY_ID', ''),
                'task_name': task.get('TASK_NAME', ''),
                'create_time': str(task.get('CREATE_TIME', '')),
                'finish_time': str(task.get('FINISH_TIME', '')) if task.get('FINISH_TIME') else None,
                'state': task.get('STATE', ''),
                'catalog': task.get('CATALOG', ''),
                'database': task.get('DATABASE', ''),
                'definition': task.get('DEFINITION', ''),  # 返回完整SQL定义
                'expire_time': str(task.get('EXPIRE_TIME', '')) if task.get('EXPIRE_TIME') else None,
                'error_code': task.get('ERROR_CODE'),
                'error_message': task.get('ERROR_MESSAGE', ''),
                'progress': task.get('PROGRESS', ''),
                'extra_message': task.get('EXTRA_MESSAGE', ''),
                'properties': task.get('PROPERTIES', ''),
                'duration_ms': duration_ms,
                'can_delete': True  # 所有任务都可以删除（使用DROP TASK命令）
            }
            formatted_tasks.append(formatted_task)
        
        logger.info(f"从task_runs表查询到 {len(formatted_tasks)} 个任务")
        return formatted_tasks
        
    except Exception as e:
        logger.error(f"获取任务列表失败: {e}")
        # 如果查询失败，返回空列表
        return []
    finally:
        if conn:
            conn.close()

@log_business_logic("get_task_names_list")
def get_task_names_list(config: Dict, search_query: str = '') -> List[str]:
    """获取StarRocks任务名称列表，支持模糊搜索"""
    conn = None
    try:
        conn = create_starrocks_connection(config)
        cursor = conn.cursor(pymysql.cursors.DictCursor)
        
        # 构建查询SQL，只获取任务名称
        base_sql = """
        SELECT DISTINCT TASK_NAME
        FROM information_schema.task_runs
        WHERE TASK_NAME IS NOT NULL AND TASK_NAME != ''
        """
        
        params = []
        
        # 如果有搜索关键词，添加模糊匹配条件
        if search_query.strip():
            base_sql += " AND TASK_NAME LIKE %s"
            params.append(f"%{search_query.strip()}%")
        
        # 按任务名称排序，限制数量
        base_sql += " ORDER BY TASK_NAME ASC LIMIT 50"
        
        logger.info(f"执行任务名称查询SQL: {base_sql}")
        cursor.execute(base_sql, params)
        results = cursor.fetchall()
        
        # 提取任务名称列表
        task_names = [row['TASK_NAME'] for row in results if row['TASK_NAME']]
        
        logger.info(f"查询到 {len(task_names)} 个任务名称")
        return task_names
        
    except Exception as e:
        logger.error(f"获取任务名称列表失败: {e}")
        return []
    finally:
        if conn:
            conn.close()

@log_business_logic("stop_running_task")
def stop_running_task(config: Dict, task_name: str, force: bool = False) -> bool:
    """停止StarRocks运行中的任务，使用DROP TASK命令"""
    conn = None
    try:
        conn = create_starrocks_connection(config)
        cursor = conn.cursor()
        
        # 直接执行DROP TASK命令
        
        # 使用DROP TASK命令停止真实任务
        # 根据StarRocks文档: DROP TASK `<task_name>` [FORCE]
        force_clause = " FORCE" if force else ""
        drop_sql = f"DROP TASK `{task_name}`{force_clause}"
        
        logger.info(f"执行停止任务命令: {drop_sql}")
        cursor.execute(drop_sql)
        conn.commit()
        
        logger.info(f"成功停止任务: {task_name}")
        return True
        
    except Exception as e:
        logger.error(f"停止任务失败 {task_name}: {e}")
        # 检查是否是因为任务已经不存在或已完成
        error_msg = str(e).lower()
        if any(keyword in error_msg for keyword in ["not found", "doesn't exist", "not exist", "unknown task"]):
            logger.info(f"任务 {task_name} 可能已经完成或不存在")
            return True
        raise e
    finally:
        if conn:
            conn.close()

# API路由
@starrocks_bp.route('/configs', methods=['GET'], endpoint='get_starrocks_configs')
@log_api_call
def get_configs():
    """获取StarRocks配置列表"""
    try:
        configs = get_starrocks_configs()
        
        # 隐藏密码信息
        for config in configs:
            if config.get('password'):
                config['password'] = '***'
        
        return jsonify(success_response(configs, "获取StarRocks配置成功"))
        
    except Exception as e:
        logger.error(f"获取StarRocks配置失败: {e}")
        return jsonify(error_response(f"获取配置失败: {str(e)}")), 500

@starrocks_bp.route('/databases', methods=['GET'], endpoint='get_starrocks_databases')
@log_api_call
def get_databases():
    """获取数据库列表"""
    try:
        config_id = request.args.get('config_id', type=int)
        if not config_id:
            return jsonify(error_response("缺少参数: config_id")), 400
        
        config = get_starrocks_config_by_id(config_id)
        databases = query_databases(config)
        
        return jsonify(success_response(databases, "获取数据库列表成功"))
        
    except Exception as e:
        logger.error(f"获取数据库列表失败: {e}")
        return jsonify(error_response(f"获取数据库列表失败: {str(e)}")), 500

@starrocks_bp.route('/tables', methods=['GET'], endpoint='get_starrocks_tables')
@log_api_call
def get_tables():
    """获取表列表"""
    try:
        config_id = request.args.get('config_id', type=int)
        database = request.args.get('database')
        
        if not config_id or not database:
            return jsonify(error_response("缺少参数: config_id, database")), 400
        
        config = get_starrocks_config_by_id(config_id)
        tables = get_table_list(config, database)
        
        return jsonify(success_response(tables, "获取表列表成功"))
        
    except Exception as e:
        logger.error(f"获取表列表失败: {e}")
        return jsonify(error_response(f"获取表列表失败: {str(e)}")), 500

@starrocks_bp.route('/transactions', methods=['GET'], endpoint='get_starrocks_transactions')
@log_api_call
def get_transactions():
    """查询运行中的事务"""
    try:
        config_id = request.args.get('config_id', type=int)
        database = request.args.get('database')
        
        if not config_id:
            return jsonify(error_response("缺少参数: config_id")), 400
        
        config = get_starrocks_config_by_id(config_id)
        transactions = query_running_transactions(config, database)
        
        return jsonify(success_response(transactions, "查询事务成功"))
        
    except Exception as e:
        logger.error(f"查询事务失败: {e}")
        return jsonify(error_response(f"查询事务失败: {str(e)}")), 500

@starrocks_bp.route('/transactions/rollback', methods=['POST'], endpoint='rollback_starrocks_transactions')
@log_api_call
def rollback_txns():
    """批量回滚事务"""
    try:
        data = request.get_json()
        config_id = data.get('config_id')
        transaction_ids = data.get('transaction_ids', [])
        database = data.get('database')
        
        if not config_id or not transaction_ids:
            return jsonify(error_response("缺少参数: config_id, transaction_ids")), 400
        
        config = get_starrocks_config_by_id(config_id)
        results = rollback_transactions(config, transaction_ids, database)
        
        message = f"回滚完成: 成功 {results['success_count']} 个, 失败 {results['failed_count']} 个"
        return jsonify(success_response(results, message))
        
    except Exception as e:
        logger.error(f"批量回滚事务失败: {e}")
        return jsonify(error_response(f"回滚事务失败: {str(e)}")), 500

@starrocks_bp.route('/transactions/show-proc', methods=['GET'], endpoint='get_transactions_by_show_proc')
@log_api_call
def get_transactions_by_show_proc():
    """使用SHOW PROC命令查询运行中的事务"""
    try:
        config_id = request.args.get('config_id', type=int)
        database = request.args.get('database')
        
        if not config_id or not database:
            return jsonify(error_response("缺少参数: config_id, database")), 400
        
        config = get_starrocks_config_by_id(config_id)
        transactions = query_running_transactions_by_show_proc(config, database)
        
        return jsonify(success_response(transactions, "查询事务成功"))
        
    except Exception as e:
        logger.error(f"通过SHOW PROC查询事务失败: {e}")
        return jsonify(error_response(f"查询事务失败: {str(e)}")), 500

@starrocks_bp.route('/transactions/filter', methods=['POST'], endpoint='filter_transactions_by_time')
@log_api_call
def filter_txns_by_time():
    """根据时间条件过滤事务"""
    try:
        data = request.get_json()
        config_id = data.get('config_id')
        database = data.get('database')
        time_condition = data.get('time_condition')
        use_show_proc = data.get('use_show_proc', False)
        
        if not config_id or not database:
            return jsonify(error_response("缺少参数: config_id, database")), 400
        
        config = get_starrocks_config_by_id(config_id)
        
        # 选择查询方式
        if use_show_proc:
            transactions = query_running_transactions_by_show_proc(config, database)
        else:
            transactions = query_running_transactions(config, database)
        
        # 应用时间过滤
        filtered_transactions = filter_transactions_by_time(transactions, time_condition)
        
        result = {
            "total_count": len(transactions),
            "filtered_count": len(filtered_transactions),
            "transactions": filtered_transactions,
            "time_condition": time_condition
        }
        
        return jsonify(success_response(result, f"过滤完成，找到 {len(filtered_transactions)} 个事务"))
        
    except Exception as e:
        logger.error(f"过滤事务失败: {e}")
        return jsonify(error_response(f"过滤事务失败: {str(e)}")), 500

@starrocks_bp.route('/transactions/rollback-by-label', methods=['POST'], endpoint='rollback_transaction_by_label')
@log_api_call
def rollback_txn_by_label():
    """根据label回滚事务"""
    try:
        data = request.get_json()
        config_id = data.get('config_id')
        label = data.get('label')
        database = data.get('database')
        
        if not config_id or not label or not database:
            return jsonify(error_response("缺少参数: config_id, label, database")), 400
        
        config = get_starrocks_config_by_id(config_id)
        result = rollback_transaction_by_label(config, label, database)
        
        if result['status'] == 'success':
            return jsonify(success_response(result, "事务回滚成功"))
        else:
            return jsonify(error_response(result['message'])), 500
        
    except Exception as e:
        logger.error(f"根据label回滚事务失败: {e}")
        return jsonify(error_response(f"回滚事务失败: {str(e)}")), 500


@starrocks_bp.route('/transactions/batch-rollback-by-labels', methods=['POST'], endpoint='batch_rollback_by_labels')
@log_api_call
def batch_rollback_by_labels():
    """批量根据labels回滚事务（兼容脚本模式）"""
    try:
        data = request.get_json()
        config_id = data.get('config_id')
        labels = data.get('labels', [])
        database = data.get('database')
        sleep_interval = data.get('sleep_interval', 0)  # 任务间隔秒数
        
        if not config_id or not labels or not database:
            return jsonify(error_response("缺少参数: config_id, labels, database")), 400
        
        config = get_starrocks_config_by_id(config_id)
        
        results = {
            "success_count": 0,
            "failed_count": 0,
            "results": []
        }
        
        for i, label in enumerate(labels):
            try:
                result = rollback_transaction_by_label(config, label, database)
                results["results"].append(result)
                
                if result['status'] == 'success':
                    results["success_count"] += 1
                else:
                    results["failed_count"] += 1
                
                # 如果不是最后一个，且设置了间隔时间，则暂停
                if i < len(labels) - 1 and sleep_interval > 0:
                    import time
                    time.sleep(sleep_interval)
                    
            except Exception as e:
                error_result = {
                    "status": "failed",
                    "message": str(e),
                    "label": label
                }
                results["results"].append(error_result)
                results["failed_count"] += 1
        
        message = f"批量回滚完成: 成功 {results['success_count']} 个, 失败 {results['failed_count']} 个"
        return jsonify(success_response(results, message))
        
    except Exception as e:
        logger.error(f"批量回滚事务失败: {e}")
        return jsonify(error_response(f"批量回滚失败: {str(e)}")), 500

@starrocks_bp.route('/query', methods=['POST'], endpoint='execute_starrocks_query')
@log_api_call
def execute_query():
    """执行SQL查询"""
    try:
        data = request.get_json()
        config_id = data.get('config_id')
        sql = data.get('sql')
        database = data.get('database')
        
        if not config_id or not sql:
            return jsonify(error_response("缺少参数: config_id, sql")), 400
        
        # 安全检查：禁止危险操作
        dangerous_keywords = ['DELETE', 'DROP', 'TRUNCATE', 'ALTER', 'CREATE', 'INSERT', 'UPDATE']
        sql_upper = sql.strip().upper()
        
        for keyword in dangerous_keywords:
            if sql_upper.startswith(keyword):
                return jsonify(error_response(f"禁止执行 {keyword} 操作")), 403
        
        config = get_starrocks_config_by_id(config_id)
        results = execute_sql_query(config, sql, database)
        
        return jsonify(success_response(results, "查询执行成功"))
        
    except Exception as e:
        logger.error(f"执行查询失败: {e}")
        return jsonify(error_response(f"查询执行失败: {str(e)}")), 500

@starrocks_bp.route('/test-connection', methods=['POST'], endpoint='test_starrocks_connection')
@log_api_call
def test_connection():
    """测试StarRocks连接"""
    conn = None
    cursor = None
    try:
        data = request.get_json()
        config_id = data.get('config_id')
        
        if not config_id:
            return jsonify(error_response("缺少参数: config_id")), 400
        
        config = get_starrocks_config_by_id(config_id)
        
        # 测试连接
        conn = create_starrocks_connection(config)
        cursor = conn.cursor()
        
        # 执行简单查询测试
        cursor.execute("SELECT 1 as test")
        result = cursor.fetchone()
        
        if result and result[0] == 1:
            return jsonify(success_response({
                "connected": True,
                "config_name": config['config_name'],
                "host": config['host'],
                "port": config['port']
            }, "连接测试成功"))
        else:
            return jsonify(error_response("连接测试失败")), 500
        
    except Exception as e:
        logger.error(f"测试连接失败: {e}")
        return jsonify(error_response(f"连接测试失败: {str(e)}")), 500
    finally:
        if cursor:
            try:
                cursor.close()
            except:
                pass
        if conn:
            try:
                conn.close()
            except:
                pass

@starrocks_bp.route('/tasks', methods=['GET'], endpoint='get_starrocks_tasks')
@log_api_call
def get_tasks():
    """获取StarRocks任务列表，支持筛选和排序"""
    try:
        config_id = request.args.get('config_id', type=int)
        state_filter = request.args.get('state', default='ALL')  # ALL, RUNNING, SUCCESS, FAILED, etc.
        sort_by = request.args.get('sort_by', default='CREATE_TIME')
        sort_order = request.args.get('sort_order', default='DESC')
        
        if not config_id:
            return jsonify(error_response("缺少参数: config_id")), 400
        
        config = get_starrocks_config_by_id(config_id)
        tasks = get_running_tasks(config, state_filter, sort_by, sort_order)
        
        # 统计任务状态
        status_counts = {}
        for task in tasks:
            state = task.get('state', 'UNKNOWN')
            status_counts[state] = status_counts.get(state, 0) + 1
        
        return jsonify(success_response({
            'tasks': tasks,
            'total_count': len(tasks),
            'status_counts': status_counts,
            'filters': {
                'state': state_filter,
                'sort_by': sort_by,
                'sort_order': sort_order
            }
        }, f"查询到 {len(tasks)} 个任务"))
        
    except Exception as e:
        logger.error(f"获取任务列表失败: {e}")
        return jsonify(error_response(f"获取任务列表失败: {str(e)}")), 500

@starrocks_bp.route('/task-names', methods=['GET'], endpoint='get_starrocks_task_names')
@log_api_call
def get_task_names():
    """获取StarRocks任务名称列表，支持模糊搜索"""
    try:
        config_id = request.args.get('config_id', type=int)
        search_query = request.args.get('search', default='')  # 搜索关键词
        
        if not config_id:
            return jsonify(error_response("缺少参数: config_id")), 400
        
        config = get_starrocks_config_by_id(config_id)
        task_names = get_task_names_list(config, search_query)
        
        return jsonify(success_response({
            'task_names': task_names,
            'total_count': len(task_names),
            'search_query': search_query
        }, f"查询到 {len(task_names)} 个任务名称"))
        
    except Exception as e:
        logger.error(f"获取任务名称列表失败: {e}")
        return jsonify(error_response(f"获取任务名称列表失败: {str(e)}")), 500

@starrocks_bp.route('/stop-task', methods=['POST'], endpoint='stop_starrocks_task')
@log_api_call
def stop_task():
    """停止StarRocks任务，使用DROP TASK命令"""
    try:
        data = request.get_json()
        config_id = data.get('config_id')
        task_name = data.get('task_name')
        force = data.get('force', False)  # 是否强制删除
        
        if not config_id or not task_name:
            return jsonify(error_response("缺少参数: config_id, task_name")), 400
        
        config = get_starrocks_config_by_id(config_id)
        success = stop_running_task(config, task_name, force)
        
        if success:
            return jsonify(success_response({
                "task_name": task_name,
                "stopped": True,
                "force": force
            }, f"任务 {task_name} 停止成功"))
        else:
            return jsonify(error_response(f"任务 {task_name} 停止失败")), 500
        
    except Exception as e:
        logger.error(f"停止任务失败: {e}")
        return jsonify(error_response(f"停止任务失败: {str(e)}")), 500

@starrocks_bp.route('/tasks/stats', methods=['GET'], endpoint='get_starrocks_tasks_stats')
@log_api_call
def get_tasks_stats():
    """获取StarRocks任务统计信息（不带LIMIT限制）"""
    try:
        config_id = request.args.get('config_id', type=int)
        
        if not config_id:
            return jsonify(error_response("缺少参数: config_id")), 400
        
        config = get_starrocks_config_by_id(config_id)
        conn = None
        
        try:
            conn = create_starrocks_connection(config)
            cursor = conn.cursor(pymysql.cursors.DictCursor)
            
            # 查询任务总数（不带LIMIT，不关心状态）
            stats_sql = "SELECT COUNT(*) as total_count FROM `_statistics_`.task_run_history"
            
            cursor.execute(stats_sql)
            result = cursor.fetchone()
            total_count = result.get('total_count', 0) if result else 0
            
            return jsonify(success_response({
                'total_count': total_count
            }, f"统计到 {total_count} 个任务"))
            
        except Exception as e:
            logger.error(f"查询任务统计失败: {e}")
            # 如果表不存在，返回0
            return jsonify(success_response({
                'total_count': 0,
                'status_counts': {}
            }, "未找到任务数据"))
        finally:
            if conn:
                conn.close()
        
    except Exception as e:
        logger.error(f"获取任务统计失败: {e}")
        return jsonify(error_response(f"获取任务统计失败: {str(e)}")), 500

@starrocks_bp.route('/clear-tasks', methods=['POST'], endpoint='clear_starrocks_tasks')
@log_api_call
def clear_tasks():
    """清空StarRocks任务历史记录（TRUNCATE task_run_history表）"""
    try:
        data = request.get_json()
        config_id = data.get('config_id')
        
        if not config_id:
            return jsonify(error_response("缺少参数: config_id")), 400
        
        config = get_starrocks_config_by_id(config_id)
        conn = None
        
        try:
            conn = create_starrocks_connection(config)
            cursor = conn.cursor(pymysql.cursors.DictCursor)
            
            logger.info(f"准备清空任务历史表: _statistics_.task_run_history")
            
            # 直接执行TRUNCATE清空任务历史表
            truncate_sql = "TRUNCATE TABLE `_statistics_`.task_run_history"
            cursor.execute(truncate_sql)
            conn.commit()
            
            logger.info(f"成功清空任务历史表")
            
            return jsonify(success_response({
                'table': '_statistics_.task_run_history'
            }, "成功清空所有任务"))
            
        except Exception as e:
            logger.error(f"清空任务失败: {e}")
            return jsonify(error_response(f"清空任务失败: {str(e)}")), 500
        finally:
            if conn:
                conn.close()
        
    except Exception as e:
        logger.error(f"清空任务失败: {e}")
        return jsonify(error_response(f"清空任务失败: {str(e)}")), 500

